package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/2
 * @Desc: 学大数据，上多易教育
 *
 *   累计到当前，各品牌中实付额最高的前10种商品
 **/
public class Job6_Top10PayProducePerBrand {

    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 订单主表映射
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        "      id BIGINT," +
                        "      status INT,                        " +
                        "      create_time timestamp(3),          " +
                        "      modify_time timestamp(3),          " +
                        "      payment_time timestamp(3),         " +
                        "      rt as modify_time        ,         " +
                        "      watermark for rt as rt - interval '0' second ,  " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order'          " +
                        ")"
        );

        // 订单详情表映射
        tenv.executeSql(
                "CREATE TABLE item_mysql (             " +
                        "      id       BIGINT,                 " +
                        "      order_id BIGINT,                 " +  // 订单id
                        "      product_id BIGINT,               " +  // 商品id
                        "      product_name STRING,             " +  // 商品名称
                        "      product_brand STRING,            " +  // 商品所属的品牌
                        "      real_amount decimal(10,2),       " +  // 商品的应付额
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order_item'     " +
                        ")"
        );

        tenv.executeSql(
                "CREATE TEMPORARY VIEW joined AS SELECT " +
                "  od.id, " +   // 订单id
                "  od.status, " +  // 订单状态
                "  od.rt, " +
                "  it.product_id, " +  // 商品id
                "  it.product_name, " +  // 商品名称
                "  it.product_brand, " +  // 商品品牌
                "  it.real_amount  " +   // 商品实付金额
                "FROM (  " +
                "   SELECT " +
                "     * " +
                "   from order_mysql " +
                "   where status in (1,2,3) and  date_format(payment_time,'yyyy-MM-dd') = CURRENT_DATE  " +
                ") od " +
                "JOIN item_mysql it " +
                "ON od.id = it.order_id");


        // 表转流
        Table joined = tenv.from("joined");
        DataStream<Row> changelogStream = tenv.toChangelogStream(joined);




        tenv.executeSql("with tmp as (\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tproduct_brand,\n" +
                "\tproduct_name,\n" +
                "\tproduct_id,\n" +
                "\tsum(real_amount) as real_amount\n" +
                "FROM  TABLE(\n" +
                "     CUMULATE(TABLE joined , DESCRIPTOR(rt), INTERVAL '1' MINUTE, INTERVAL '24' HOUR  )\n" +
                ")\n" +
                "GROUP BY \n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tproduct_brand,\n" +
                "\tproduct_name,\n" +
                "\tproduct_id\n" +
                ")\t\n" +
                "\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tproduct_brand,\n" +
                "\tproduct_name,\n" +
                "\tproduct_id,\n" +
                "\treal_amount,\n" +
                "\trn\n" +
                "FROM (\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tproduct_brand,\n" +
                "\tproduct_name,\n" +
                "\tproduct_id,\n" +
                "\treal_amount,\n" +
                "\trow_number() over(partition by window_start,window_end,product_brand order by real_amount desc) as rn\n" +
                "FROM tmp\n" +
                ") o\n" +
                "WHERE rn<=10").print();





    }

}
